package com.niit.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark_RDD_Transform01 {

  def main(args: Array[String]): Unit = {
    //准备环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark")
    val sc = new SparkContext(sparkConf);
    /*
      map:将处理的数据逐条进行映射转换，这里的转换可以是类型的转换，也可以是值的转换。
      需求：将列表中 1 2 3 4 转为 2 4 6 8
     */
    val rdd1 = sc.makeRDD(List(1,2,3,4))
    val mapRdd1 =  rdd1.map(_*2) //一个rdd转换后 会生成一个新的rdd
    mapRdd1.collect().foreach(println)

    /*
    mapPartitions:将待处理的数据以分区为单位发送到计算节点进行处理，这里的处理是指可以进行任意的处
理，哪怕是过滤数据。
      需求：找个每个分区的最大值
     */
    val rrd2 = sc.makeRDD(List(1,2,3,4),2) //【1，2】  【3，4】
   val mapRdd2 = rrd2.mapPartitions(iter => {
      //iter 指的是每个分区的数据 【1，2】  【3，4】
      List(iter.max).iterator
    })
    mapRdd2.collect().foreach(println)
    /*
    flatMap:将处理的数据进行扁平化后再进行映射处理，所以算子也称之为扁平映射
     */
    val rdd3 =  sc.makeRDD(List( List(1,2) , List(3,4)  ))//==>(1,2,3,4)
     val mapRdd3 =  rdd3.flatMap( list =>{
       list
     } )
    mapRdd3.collect().foreach(println)
    //需求：以空格拆分单词
    val rdd4 = sc.makeRDD( List("Hello Spark" ,"Hello Hadoop")) //==> (Hello,Spark,Hello,Hadoop)
    val mapRdd4 = rdd4.flatMap(s =>{
      s.split(" ")  // ( (Hello),(Spark),(Hello),(Hadoop) ) ==> (Hello,Spark,Hello,Hadoop)
    })
    mapRdd4.collect().foreach(println)

    /*
    glom: 将同一个分区的数据直接转换为相同类型的内存数组进行处理，分区不变
     */
    val rdd5 = sc.makeRDD(List(1,2,3,4),2)//【1，2】 【3，4】
    val glomRdd1:RDD[Array[Int]] = rdd5.glom()//[[1,2][3,4]]
    glomRdd1.collect().foreach(data => println(data.mkString(",") ))
    //需求：获得两个分区最大值的和  6
    val rdd6 = sc.makeRDD(List(1,2,3,4),2)//【1，2】 【3，4】
    val gloRdd2 =  rdd6.glom()
    val maxRdd =  gloRdd2.map(array=>{//[1,2] => 2      [3,4] =>4
      array.max
    })
    println(maxRdd.collect().sum)

    /*
    groupBy:将数据根据指定的规则进行分组, 分区默认不变，但是数据会被打乱重新组合，我们将这样
  的操作称之为 shuffle。极限情况下，数据可能被分在同一个分区中 一个组的数据在一个分
  区中，但是并不是说一个分区中只有一个组
       */
    val rdd7 = sc.makeRDD(List(1,2,3,4),2) //奇数一组  偶数一组
    //先去自定义一个分组的方法
    def groupFunction(num:Int): Int ={
      num % 2
    }
    val groupRdd:RDD[(Int,Iterable[Int])] =  rdd7.groupBy(groupFunction)
    groupRdd.collect().foreach(println)

    val groupRdd2 =  rdd7.groupBy(_%2)
    groupRdd2.collect().foreach(println)
    //需求 ：按照单词的首字母进行分组
    val rdd8 = sc.makeRDD(List("Hello","Spark","HBase","Scala","Kafka"))
    val groupRdd3 = rdd8.groupBy(_.charAt(0))// Hello=>H    Spark=>S   HBase=>H   Scala=>S    Kafka=>K
    groupRdd3.collect().foreach(println)

  }

}
